package com.gitee.zhangchenyan.takin.event.producer;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Properties;
import java.util.concurrent.Future;


/**
 * 事件总线生产者
 *
 * @author zl
 * @date 2022/09/18
 */
@Slf4j
@Component
public class EventBusProducer implements IEventBusProducer {

    private final EventBusProducerConfig eventBusProducerConfig;

    private boolean isInit = false;
    private KafkaProducer<String, String> kafkaProducer = null;

    public EventBusProducer(EventBusProducerConfig eventBusProducerConfig) {
        this.eventBusProducerConfig = eventBusProducerConfig;
        initKafkaProducer();
    }


    /**
     * 发布
     *
     * @param topic 主题
     * @param data  数据
     */
    @SneakyThrows
    @Override
    public void publish(String topic, String data) {
        isInit();
        ProducerRecord<String, String> kafkaMessage = new ProducerRecord<>(topic, data);
        Future<RecordMetadata> metadataFuture = kafkaProducer.send(kafkaMessage, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e != null) {
                    log.error("EventBus publish Exception : " + e);
                }
            }
        });
        RecordMetadata recordMetadata = metadataFuture.get();
        log.info("EventBus publish data [{}] result [{}]", data, recordMetadata.toString());
    }


    /**
     * 是否初始化
     */
    private void isInit() {
        if (!isInit) {
            throw new IllegalArgumentException("Kafka Not Connect ,Checking KafkaProducer Configuration Parameters");
        }
    }


    /**
     * init卡夫卡生产商
     */
    private void initKafkaProducer() {
        try {
            if (StringUtils.hasLength(eventBusProducerConfig.getBootstrapServers())) {
                Properties props = new Properties();
                //设置接入点，请通过控制台获取对应Topic的接入点
                props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, eventBusProducerConfig.getBootstrapServers());
                //Kafka消息的序列化方式
                props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
                props.put(ProducerConfig.RETRIES_CONFIG, eventBusProducerConfig.getRetries());
                props.put(ProducerConfig.BATCH_SIZE_CONFIG, eventBusProducerConfig.getBatchSize());
                props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, eventBusProducerConfig.getBufferMemory());
                props.put(ProducerConfig.ACKS_CONFIG, eventBusProducerConfig.getAcks());
                //构造Producer对象，注意，该对象是线程安全的，一般来说，一个进程内一个Producer对象即可；
                //如果想提高性能，可以多构造几个对象，但不要太多，最好不要超过5个
                kafkaProducer = new KafkaProducer<>(props);
                isInit = true;
            }
        } catch (Exception ex) {
            log.error("Kafka failed to connect : " + ex);
        }

    }
}
